package com.xtianzhuang.www.study2019.kafka;


import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaComsumer {



	public static void getConsumer(String targetTopic,
			String targetServerIp) throws IOException {
		Map<String, Object> props = new HashMap<String, Object>();
		props.put("enable.auto.commit", "true");
		props.put("auto.offset.reset", "earliest");
		props.put("auto.commit.interval.ms", "5000");
		props.put("group.id", "20190823test2");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		String[] topics = new String[]{targetTopic};
		props.put("bootstrap.servers", targetServerIp);
		KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props);
		// 自动分配分区
		consumer.subscribe(Arrays.asList(topics));
		ConsumerRecords<byte[], byte[]> records = consumer.poll(5000);
		if (records.count() > 0) {
			System.out.printf("count = {}", records.count());
		} else {
			System.out.println("count =0");
		}
		for (ConsumerRecord<byte[], byte[]> record : records) {
			System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
	}

	}


	public static void main(String[] args) throws IOException {
		KafkaComsumer.getConsumer(
				"CRM_CUE.UCR_CEN.TD_B_PACKAGE",
				"10.161.11.186:6667,10.161.11.187:6667,10.161.11.188:6667,10.161.11.189:6667,10.161.11.190:6667,10.161.11.191:6667,10.161.11.192:6667,10.161.11.193:6667,10.161.11.194:6667,10.161.11.195:6667");
	}
}
